package com.dongzili.common.disruptor.service;

import cn.hutool.core.lang.UUID;
import com.dongzili.common.disruptor.entity.MessageEvent;
import com.dongzili.common.disruptor.enums.MessageType;
import com.lmax.disruptor.RingBuffer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import jakarta.annotation.Resource;

@Slf4j
@Service
public class DisruptorMqService<T>{

    @Resource
    RingBuffer<MessageEvent<T>> ringBuffer;
    @Resource
    RedisTemplate<String, MessageEvent<Object>> redisTemplate;
    public static final String redisKey = "disruptor:message:";

    public void sendMq(MessageType type, MessageEvent<T> messageEvent) {
        //获取下一个Event槽的下标
        long sequence = ringBuffer.next();
        String id = UUID.fastUUID().toString();
        try {
            //给Event填充数据
            MessageEvent<T> event = ringBuffer.get(sequence);
            event.setValue(messageEvent.getValue());
            event.setSequence(messageEvent.getSequence());
            event.setData(messageEvent.getData());
            event.setId(id);
            event.setType(type);
            redisTemplate.opsForHash().put(redisKey, id, event);
//            log.info("往消息队列中添加消息：{}", event);
        } catch (Exception e) {
            log.error("向messageModelRingBuffer添加事件失败 : e = {}",e.getMessage(),e);
            redisTemplate.opsForHash().delete(redisKey, id);
        } finally {
            //发布Event，激活观察者去消费，将sequence传递给改消费者
            //注意最后的publish方法必须放在finally中以确保必须得到调用；如果某个请求的sequence未被提交将会堵塞后续的发布操作或者其他的producer
            ringBuffer.publish(sequence);
        }
    }
}
